package com.platform.apex.output;

import java.util.Map;

import org.elasticsearch.action.index.IndexRequestBuilder;

import com.datatorrent.api.Operator;
import com.datatorrent.api.Context.OperatorContext;
import com.datatorrent.contrib.elasticsearch.AbstractElasticSearchOutputOperator;

/**
 * Concrete implementation of {@link AbstractElasticSearchOutputOperator}
 * demonstrating the functionality for Tuples of Map type.
 *
 * @category tjsj
 *
 * @since 2.1.0
 */
public class ElasticAppSearchMapOutputOperator<T extends Map<String, Object>>
		extends AbstractAppElasticSearchOutputOperator<T, ElasticAppSearchConnectable>{

	private String idField;
	private String indexName;
	private String type;

	

	/**
	 *
	 */
	public ElasticAppSearchMapOutputOperator() {
		this.store = new ElasticAppSearchConnectable();
	}

	@Override
	public void beginWindow(long windowId) {
		super.beginWindow(windowId);
		
//		boolean indexExist = checkIndexExists(indexName);
//		if (!indexExist) {
//			createIndexType(indexName, type);
//		}
		
	}

	
	@Override
	public void endWindow() {
		// TODO Auto-generated method stub
		super.endWindow();
		
	}
	// check index is exist
	

	@Override
	public void setup(OperatorContext context) {
		// TODO Auto-generated method stub
		super.setup(context);
	}
	/**
	 * mapping 一旦定义，之后就不能修改。
	 * 
	 * @return
	 * @throws Exception
	 */
	
	
	/*
	 * (non-Javadoc)
	 *
	 * @see
	 * com.datatorrent.contrib.elasticsearch.AbstractElasticSearchOutputOperator
	 * #setSource(org.elasticsearch.action.index .IndexRequestBuilder,
	 * java.lang.Object)
	 */
	@Override
	protected IndexRequestBuilder setSource(IndexRequestBuilder indexRequestBuilder, T tuple) {
		return indexRequestBuilder.setSource(tuple);
	}

	/*
	 * (non-Javadoc)
	 *
	 * @see
	 * com.datatorrent.contrib.elasticsearch.AbstractElasticSearchOutputOperator
	 * #getId(java.lang.Object)
	 */
	@Override
	public String getId(T tuple) {
		if (idField == null) {
			return null;
		} else {
			return tuple.get(idField).toString();
		}

	}

	/**
	 * @param idField
	 *            the idField to set
	 */
	public void setIdField(String idField) {
		this.idField = idField;
	}

	/**
	 * @return the idField
	 */
	public String getIdField() {
		return idField;
	}

	/**
	 * @param indexName
	 *            the indexName to set
	 */
	public void setIndexName(String indexName) {
		this.indexName = indexName;
	}

	/*
	 * (non-Javadoc)
	 *
	 * @see
	 * com.datatorrent.contrib.elasticsearch.AbstractElasticSearchOutputOperator
	 * #getIndexName(java.lang.Object)
	 */
	@Override
	protected String getIndexName(T tuple) {
		return indexName;
	}

	public String getIndexName() {
		return indexName;
	}

	/**
	 * @param type
	 *            the type to set
	 */
	public void setType(String type) {
		this.type = type;
	}

	/*
	 * (non-Javadoc)
	 * 
	 * @see
	 * com.datatorrent.contrib.elasticsearch.AbstractElasticSearchOutputOperator
	 * #getType(java.lang.Object)
	 */
	@Override
	protected String getType(T tuple) {
		return type;
	}

	public String getType() {
		return type;
	}

}
